package org.example.operator.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * rdd的算子，也就是rdd的方法
 *
 * rdd的map的buff版本
 */
object Spark02_RDD_Operator_MapPartitions {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("RDD")
    val context = new SparkContext(conf)

    val rdd: RDD[Int] = context.makeRDD(List(1,2,3,4),2)

    // map的版本效率比较低，类似字节读取，而map还有更高效的方式
    // 他会一次多读取，最后一起处理
    // 迭代器会选择将 某个分区的全部内容一次性读取到内存中处理
    // 但是 mapPartitions 存在缺点，由于一次读取，所以使用完后，内存不会释放，容易溢出
    rdd.mapPartitions(iter=>{
      // 有多少个分区，这里会执行多少次
      iter.map(_*2)
    })

    // 得到每个分区的最大值
    rdd.mapPartitions(iter=>{
        List(iter.max).iterator
    })

    rdd.collect().foreach(println)

    context.stop();  }
}
